package com.dyrnq.seckill.mq.disruptor;

import cn.hutool.core.thread.ThreadFactoryBuilder;
import com.dyrnq.seckill.mq.MessageVo;
import com.dyrnq.seckill.service.impl.SeckillOrderService;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.dsl.Disruptor;
import lombok.RequiredArgsConstructor;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.ThreadFactory;

@Configuration
@RequiredArgsConstructor
@ConditionalOnProperty(value = "queue.impl", havingValue = "disruptor")
public class DisruptorConfig {
    private final DisruptorProperties properties;
    private final SeckillOrderService seckillOrderService;

    @Bean(initMethod = "start", destroyMethod = "shutdown")
    public Disruptor<MessageVo> disruptor() {
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNamePrefix("disruptor-").build();
        EventFactory<MessageVo> eventFactory = MessageVo::new;
        Disruptor<MessageVo> disruptor = new Disruptor<>(eventFactory, properties.getRingBufferSize(), namedThreadFactory);
        DisruptorConsumer[] consumers = new DisruptorConsumer[10];
        for (int i = 0; i < consumers.length; i++) {
            consumers[i] = new DisruptorConsumer(seckillOrderService);
        }
        disruptor.handleEventsWithWorkerPool(consumers);
        return disruptor;
    }


    @Bean
    public DisruptorProducer producer(Disruptor<MessageVo> disruptor) {
        return new DisruptorProducer(disruptor.getRingBuffer());
    }
}
